feat(write): route postpone batch writes to fixed buckets - #659
feat(write): route postpone batch writes to fixed buckets#659XiaoHongbo-Hope wants to merge 6 commits into
Conversation
|
|
|
Thanks @JingsongLi, fixed in a18eede. |
|
| /// Planning state for batch writes to postpone-bucket tables. Partitions with | ||
| /// an existing real-bucket count are written incrementally, while only new | ||
| /// partitions are buffered until their bucket count can be inferred. | ||
| struct PostponeFixedBucketState { |
There was a problem hiding this comment.
Create a new rs file for postpone writer.
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
I found three behavior gaps compared with the merged PyPaimon implementation: distributed writers do not share a bucket plan, overwrite rescaling is rejected by conflict detection, and the size target is validated even when the row-count target should take precedence. Details are inline.
| .copied() | ||
| .unwrap_or(0) | ||
| }; | ||
| let total_buckets = infer_bucket_count( |
There was a problem hiding this comment.
Each TableWrite derives total_buckets from only its local buffered rows. However, the C API supports merging messages from multiple fixed-bucket writers that share a commit_user, and there is no way to provide those writers with one precomputed bucket plan. Two workers writing the same new partition can therefore infer different counts and make the whole commit fail; even when they infer the same count, the plan is based on shard-local rather than global batch statistics and can under-bucket the partition. PyPaimon avoids this by aggregating partition statistics on the driver and injecting the same PostponeBucketPlan into every worker. Please expose and validate a precomputed partition -> total_buckets plan in the Rust builder/writer and C API, or explicitly reject this multi-writer mode.
| check_from_snapshot: Option<i64>, | ||
| ) -> Result<()> { | ||
| self.check_delete_entries_against_base(base_entries, delta_entries)?; | ||
| self.check_total_bucket_conflicts(base_entries, delta_entries)?; |
There was a problem hiding this comment.
This bucket-count check runs unconditionally before considering commit_kind, and it compares raw base and delta entries. An overwrite contains DELETE entries for the old layout followed by ADD entries for the replacement layout, so a valid rescale such as 1 -> 3 buckets is rejected as a conflict. The upstream Java implementation skips the old-layout consistency check for OVERWRITE, and PyPaimon has test_postpone_overwrite_allows_bucket_rescale. Please continue checking that all new ADD entries in the delta agree, but compare them with the base layout only for non-overwrite commits (or apply ADD/DELETE changes before checking the final active entries).
| bucket_function_type, | ||
| max_parallelism: options.postpone_batch_write_fixed_bucket_max_parallelism()?, | ||
| target_rows_per_bucket: options.postpone_target_row_num_per_bucket()?, | ||
| target_size_per_bucket: options.postpone_target_size_per_bucket()?, |
There was a problem hiding this comment.
postpone.target-size-per-bucket is parsed and validated even when postpone.target-row-num-per-bucket is configured. The option contract says that the size target is ignored in that case, and the PyPaimon planner only reads it in the row target is None branch. With a valid row target plus an invalid or zero size target, Rust currently rejects writer creation even though the size value is unused. Please parse and validate the size target only when no row-count target is present.
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
One new regression remains after the latest fixes: the core builder now auto-enables fixed mode without the global plan that PyPaimon obtains at the integration layer. Details inline. The previous three findings are fixed.
| pub fn new(table: &'a Table) -> Self { | ||
| let schema = table.schema(); | ||
| let options = CoreOptions::new(schema.options()); | ||
| let postpone_fixed_bucket = options.bucket() == POSTPONE_BUCKET |
There was a problem hiding this comment.
Why enabling fixed_bucket write by default? Can you create a real PostponeFixedBucketWriteBuilder?
2 participants
Purpose
Align Rust batch writes for postpone-bucket primary-key tables with the fixed-bucket behavior proposed in apache/paimon#8985. This extracts the core writer change from #658; Go binding and release changes remain in the draft PR.
Changes
postpone.batch-write-fixed-bucket=falsetotal_bucketsin commit messages and reject concurrent conflicting bucket-count decisionsVerification
cargo fmt --all -- --checkcargo check --locked --offline -p paimoncargo test --locked --offline -p paimon test_postpone_ --libcargo test --locked --offline -p paimon-datafusion test_postpone_batch_write_uses_visible_fixed_bucket --test pk_tables -- --exact